package com.qiall.gateway.controller.websocket;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class DataKafkaQueue {
    private static final String TOPIC = "gate_way_web_socket_message";
    private static final String TOPIC_RECEIVE = "gate_way_web_socket_receive";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private SessionManage sessionManage;

    public void sedSocketMessage(Protocol protocol) {
        kafkaTemplate.send(TOPIC, protocol.getSessionId(), JSON.toJSONString(protocol));
    }

    @KafkaListener(topics = {TOPIC_RECEIVE}, concurrency = "6")
    public void kafkaListener(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String recordValue = record.value();
        Protocol protocol = JSON.parseObject(recordValue, Protocol.class);
        sessionManage.sendMessage(protocol);
        ack.acknowledge();
    }
}
